package org.eventime

import org.FlinkStreamApp
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * description ：窗口等待迟到元素
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/23 12:22
 * modified By ：
 * version:    : 1.0
 */
object AllowedLatenessExample extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    val stream = environment.socketTextStream("hadoop101", 9999, '\n')
    stream.map(
      l => {
        val splits = l.split("\\s")
        (splits(0), splits(1).toLong * 1000)
      }
    )
      .assignTimestampsAndWatermarks(
        // 设置最大延迟时间5s
        new BoundedOutOfOrdernessTimestampExtractor[(String, Long)](Time.seconds(5)) {
          override def extractTimestamp(element: (String, Long)) = element._2
        }
      )
      .keyBy(_._1)
      .timeWindow(Time.seconds(5))
      .process(new UpdatingWindowCountFunction())
      .print()
  }

  // [IN, OUT, KEY, W <: Window]
  class UpdatingWindowCountFunction extends ProcessWindowFunction[(String, Long), String, String, TimeWindow] {
    // 设置一个状态保存， 是否第一触发了窗口关闭，用来区分是否迟到元素，而窗口是等待情况下的
    lazy val isFirstCloseWindow = getRuntimeContext.getState(
      new ValueStateDescriptor[Boolean]("isFirstCloseWindow", Types.of[Boolean])
    )

    override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[String]) = {
      // 获取元素总数
      val cnt = elements.size

      // 如果第一次窗口关闭之后，再进来的，就是后面等待中的窗口，这里会进来多次（迟到的元素，取决于你的窗口最大的等待时间）
      if (isFirstCloseWindow.value()) {
        // 处理迟到元素
        out.collect("迟到元素来了！总数为： " + cnt.toString)
      } else {
        // 正常元素
        out.collect("第一次窗口求值，总数为： " + cnt.toString)

        // 第一次窗口关闭之后，首先将 isFirstCloseWindow 设置为true
        isFirstCloseWindow.update(true)
      }
    }
  }

}
